package cn.smileyan.demos.functions;

import cn.smileyan.demos.entity.MergedData;
import cn.smileyan.demos.entity.PieceData;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;
import java.util.LinkedList;

@Slf4j
public class MergeProcessingFunction extends ProcessWindowFunction<PieceData, MergedData, String, TimeWindow> {
    @Override
    public void process(String s,
                        ProcessWindowFunction<PieceData, MergedData, String, TimeWindow>.Context context,
                        Iterable<PieceData> elements,
                        Collector<MergedData> out) throws Exception {
        log.info("[{}] starting merge process", s);
        MergedData mergedData = new MergedData();

        Iterator<PieceData> iterator = elements.iterator();
        PieceData first = iterator.next();
        mergedData.setId(first.getId());
        mergedData.setSize(first.getSize());

        LinkedList<Tuple2<Integer, Double>> values = new LinkedList<>();
        while (iterator.hasNext()) {
            PieceData pieceData = iterator.next();
            values.add(Tuple2.of(pieceData.getIndex(), pieceData.getValue()));
        }
        mergedData.setValues(values);

        out.collect(mergedData);
    }
}
